package com.eastfair.transaction.mq.service;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.eastfair.cache.projectcore.DictionaryCacheKeyBuilder;
import com.eastfair.cache.repository.CachePlusOps;
import com.eastfair.core.model.CacheHashKey;
import com.eastfair.core.rocketmq.MqCacheKeyBuilder;
import com.eastfair.mq.model.MySimpleMessage;
import com.eastfair.mq.util.MessageExtUtil;
import com.eastfair.mq.util.MessageKeyUtil;
import com.eastfair.transaction.util.TransactionUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;

import java.util.HashMap;
import java.util.Map;

@Slf4j
public abstract class AbstractBaseTransactionService<T> implements BaseTransactionService<T> {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Value("${rocketmq.consumer.transactionTopic}")
    private String transactionTopic;

    @Autowired
    private CachePlusOps cachePlusOps;

    @Autowired
    private TransactionUtil transactionUtil;

    @Autowired
    private MessageExtUtil messageExtUtil;

    /**
     * 发送队列，预发送，发送后执行本地方法，之后在commit发送
     */
    public void sendQueueMessage(Object object){
        //组装MySimpleMessage
        MySimpleMessage mySimpleMessage = transactionUtil.assembleSimpleMessage(object);
//        Message message = MessageBuilder.withPayload(JSONObject.toJSONString(mySimpleMessage))
//                .build();
        MessageHeaders messageHeaders = new MessageHeaders(mySimpleMessage.getHeaders());
//        Message message = MessageBuilder.createMessage(mySimpleMessage,messageHeaders);

//        Message message = MessageBuilder.withPayload(JSONObject.toJSONString(mySimpleMessage))
//                .build();

        //消息对象生成
        Message<String> message = MessageBuilder.withPayload(JSONObject.toJSONString(mySimpleMessage))
                .setHeader(RocketMQHeaders.TAGS, messageHeaders.get("tag"))
                .setHeader(RocketMQHeaders.KEYS, messageHeaders.get("key"))
                .build();

        //唯一业务key 用于幂等
        CacheHashKey typeKey = new MqCacheKeyBuilder().hashKey(mySimpleMessage.getHeaders().get("key").toString());
        cachePlusOps.set(typeKey, 1);
        rocketMQTemplate.sendMessageInTransaction(transactionTopic,
                message, null);
    }

    /**
     * 组装header头数据
     * @param message
     */
    public void assembleContextData(MySimpleMessage message){
        messageExtUtil.parentContextData(message.getHeaders());
    }

}
